package com.ibm.datamodel.timeseries.parser;

import java.io.IOException;
import java.text.ParseException;
import java.util.Arrays;
import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.interfaces.IMessageParser;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.DataTypes;
import com.ibm.util.DateParser;
import com.ibm.util.exception.CPSException;

import io.protostuff.Input;
import io.protostuff.Output;
import io.protostuff.Schema;

/**
 * @author Yaoliang Chen
 * @version Nov 20, 2015 
 */
abstract public class ProtostuffMessageParser implements Schema<ParsedDataMessage>, IMessageParser {
	private DateParser dateParser = new DateParser();
	protected ITimeSeriesSchema schema;
	private byte[] dataTypes;
	private String[] tagNames;
	static private Logger logger = LoggerFactory.getLogger(ProtostuffMessageParser.class);

	public ProtostuffMessageParser(ITimeSeriesSchema schema) throws CPSException {
		this.schema = schema;
		this.dataTypes = schema.getDataTypes();
		this.tagNames = schema.getTagNames();
	}

	public boolean isInitialized(ParsedDataMessage message) {
		return message.getTsKey() != null;
	}

	@Override
	public void mergeFrom(Input input, ParsedDataMessage message) throws IOException {
		// try {
		if (message.getTimestamp() == null) {
			long time = new Date().getTime();
			message.setValue(this.schema.getTimestampIndex(), time);
			message.setTimestamp(time);
		}

		while (true) {
			int fieldIndex = input.readFieldNumber(this) - 1;
			if (fieldIndex == -2) {
				/*
				 * throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
				 * "Please provide the message according to schema " +
				 * Arrays.toString(tagNames));
				 */
				// input.readByteArray();
				continue;
			} else if (fieldIndex < 0) {
				break;
			}

			if (fieldIndex < 0 || fieldIndex >= dataTypes.length) {
				input.handleUnknownField(fieldIndex + 1, this);
			}

			switch (this.dataTypes[fieldIndex]) {
			case DataTypes.DT_LONG:
				long value;
				value = input.readInt64();
				message.setValue(fieldIndex, value);
				if (fieldIndex == this.schema.getTimestampIndex()) {
					message.setTimestamp(value);
				} else if (fieldIndex == this.schema.getKeyIndex()) {
					message.setTsKey(value);
				}
				break;
			case DataTypes.DT_STRING:
				String str = input.readString();
				if (fieldIndex == this.schema.getTimestampIndex()) {
					try {
						long ts = dateParser.getTimestamp(str);
						message.setValue(fieldIndex, ts);
						message.setTimestamp(ts);
					} catch (ParseException e) {
						throw new IOException(e);
					}

				} else {
					message.setValue(fieldIndex, str);
				}
				break;
			case DataTypes.DT_DOUBLE:
				message.setValue(fieldIndex, input.readDouble());
				break;
			case DataTypes.DT_BOOLEAN:
				message.setValue(fieldIndex, input.readBool());
				break;
			default:
				input.handleUnknownField(fieldIndex, this);
			}
		}
		// } catch (CPSException e) {
		// logger.error(Throwables.getStackTraceAsString(e));
		// logger.error(e.getMessage());
		// }
	}

	public void writeTo(Output output, ParsedDataMessage message) throws IOException {
		int len = message.getValueCount();
		for (int i = 0; i < len; ++i) {
			if (DataTypes.isArray(schema.getDataType(i))) {
				if (message.getArrayValue(i) == null) {
					logger.debug("Value in index " + i + " is null.");
					output.writeString(i + 1, null, false);
					continue;
				}
			} else {
				if (message.getValue(i) == null) {
					logger.debug("Value in index " + i + " is null.");
					output.writeString(i + 1, null, false);
					continue;
				}
			}
			switch (this.dataTypes[i]) {
			case DataTypes.DT_LONG:
				output.writeInt64(i + 1, (Long) message.getValue(i), false);
				break;
			case DataTypes.DT_STRING:
				String str = null;
				if (i == this.schema.getTimestampIndex()) {
					long tstamp = (Long) message.getValue(i);
					str = dateParser.getTimestampString(tstamp);
				} else {
					str = (String) message.getValue(i);
				}
				output.writeString(i + 1, str, false);
				break;
			case DataTypes.DT_DOUBLE:
				output.writeDouble(i + 1, (Double) message.getValue(i), false);
				break;
			case DataTypes.DT_BOOLEAN:
				output.writeBool(i + 1, (Boolean) message.getValue(i), false);
				break;
			case DataTypes.DT_DOUBLE_ARRAY:
			case DataTypes.DT_INT_ARRAY:
			case DataTypes.DT_STRING_ARRAY:
				String tmp = Arrays.toString(message.getArrayValue(i));
				output.writeString(i + 1, tmp, false);
				break;
			default:
			}

		}

	}

	public ParsedDataMessage newMessage() {
		return schema.newMessage();
	}

	public Class<ParsedDataMessage> typeClass() {
		return ParsedDataMessage.class;
	}

	public String messageName() {
		return ParsedDataMessage.class.getSimpleName();
	}

	public String messageFullName() {
		return ParsedDataMessage.class.getName();
	}

	// the mapping between the field names to the field numbers.
	@Override
	public String getFieldName(int number) {
		if (number <= 0 || number > this.schema.getTagCount()) {
			return null;
		}
		return tagNames[number - 1];
	}

	@Override
	public int getFieldNumber(String name) {
		Integer number = schema.getTagId(name);
		return number == null ? -1 : number.intValue() + 1;
	}
	
	@Override
	public ITimeSeriesSchema getSchema() {
		return schema;
	}

	
}
